package DianShang_2024.ds_01.clean

import org.apache.spark.sql.SparkSession

object clean01 {
  def main(args: Array[String]): Unit = {
    /*
        抽取ods库中user_info表中昨天的分区（子任务一生成的分区）数据，并结合dim_user_info最新分区现有的数据，根据id合并数据到dwd库
        中dim_user_info的分区表（合并是指对dwd层数据进行插入或修改，需修改的数据以id为合并字段，根据operate_time排序取最新的一条），
        分区字段为etl_date且值与ods库的相对应表该值相等，同时若operate_time为空，则用create_time填充，并添加dwd_insert_user、
        dwd_insert_time、dwd_modify_user、dwd_modify_time四列,其中dwd_insert_user、dwd_modify_user均填写“user1”。若该条记录第一次
        进入数仓dwd层则dwd_insert_time、dwd_modify_time均存当前操作时间，并进行数据类型转换。若该数据在进入dwd层时发生了合并修改，
        则dwd_insert_time时间不变，dwd_modify_time存当前操作时间，其余列存最新的值。使用hive cli执行show partitions dwd.dim_user_info命令
     */

    //  准备sparksql的环境
      val spark=SparkSession.builder()
        .master("local[*]")
        .appName("数据清洗第一题")
        .enableHiveSupport()
        .getOrCreate()

    spark.sql("use dwd")


    /*
          处理逻辑:
          1.首先对dwd.dim_user_info和ods.user_info2表里面的operate_time是否为空进行判断，如果为空的话就用create_time填充,并且在判断的
          时候ods.user_info2表就将题目要求创建的字段创建好
          2.将两个表进行union all(不会去除重复值)合并。(如果使用union连接的话会去除重复行)
          3.使用排名函数(窗口函数)将id相同的数据按照operate_time进行排序，然后只取第一行的数据，这样就算修改完数据了
     */

    //  t1:对dwd.dim_user_info里面的operate_time是否为空进行判断，为空的话使用create_time填充
    spark.sql(
      """
        |select
        |id,
        |login_name,
        |nick_name,
        |passwd,
        |name,
        |phone_num,
        |email,
        |head_img,
        |user_level,
        |birthday,
        |gender,
        |create_time,
        |if(operate_time is not null, operate_time,create_time)  as  operate_time,
        |dwd_insert_user,
        |dwd_insert_time,
        |dwd_modify_user,
        |dwd_modify_time
        |from
        |dwd.dim_user_info
        |""".stripMargin).createOrReplaceTempView("t1")

    //  t2:判断ods.user_info2表里面的operate_time是否为空，是的话用create_time填充，并且创建题目要求的四个字段
    /*
        1. "user1"   as dwd_insert_user,   表示给字段赋值，并且给字段取一个别名为dwd_insert_user
        2.to_timestamp(）:按照指定格式将字符串转化为时间戳类型
                  date_format():按照指定格式将时间转化为字符串
     */

    spark.sql(
      """
        |select
        |id,
        |login_name,
        |nick_name,
        |passwd,
        |name,
        |phone_num,
        |email,
        |head_img,
        |user_level,
        |birthday,
        |gender,
        |create_time,
        |if(operate_time is not null,operate_time,create_time) as operate_time,
        |'user1' as dwd_insert_user,
        |to_timestamp(date_format(current_timestamp(),'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd HH:mm:ss') as dwd_insert_time,
        |'user1' as dwd_modify_user,
        |to_timestamp(date_format(current_timestamp(),'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd HH:mm:ss') as dwd_modify_time
        |from
        |ods.user_info2
        |where etl_date='20231017'
        |""".stripMargin).createOrReplaceTempView("t2")

    //  t3:将t1和t2使用union all合并得到t3
    spark.sql(
      """
        |select * from t1
        |union all
        |select * from t2
        |""".stripMargin).createOrReplaceTempView("t3")

    //  t4:使用排名函数对相同id的数据进行窗口排序,由于窗口函数生成的列名，不能直接当作过滤条件，所以需要在外层才可以当作过滤条件，所以得到t4方便
    //  后面使用
    /*
        min(dwd_insert_time) over(partition by id order by operate_time desc)  as dwd_insert_time,:在每个id分组内，
        根据operate_time字段的降序排列，计算该分组内最小的dwd_insert_time值,然后这一列的数据都是用的这一个

     */

    spark.sql(
      """
        |select
        |id,
        |login_name,
        |nick_name,
        |passwd,
        |name,
        |phone_num,
        |email,
        |head_img,
        |user_level,
        |birthday,
        |gender,
        |create_time,
        |operate_time,
        |row_number() over(partition by id order by operate_time desc)   as  row_number,
        |dwd_insert_user,
        | min(dwd_insert_time) over(partition by id order by operate_time desc)  as dwd_insert_time,
        |dwd_modify_user,
        |dwd_modify_time
        |from
        |t3
        |""".stripMargin).createOrReplaceTempView("t4")

    //  t5:根据t4得到每个窗口里面第一列的数据,(窗口函数创建的列名只能在外层才可以当作过滤条件)
    spark.sql("select * from t4 where row_number=1").createOrReplaceTempView("t5")
    spark.sql("select * from t5").show

    //  将数据导入到dwd.dim_user_info
    spark.sql(
      """
        |insert into table dwd.dim_user_info
        |partition(etl_date='20231017')
        |select
        |id,
        |login_name,
        |nick_name,
        |passwd,
        |name,
        |phone_num,
        |email,
        |head_img,
        |user_level,
        |birthday,
        |gender,
        |create_time,
        |operate_time,
        |dwd_insert_user,
        |dwd_insert_time,
        |dwd_modify_user,
        |dwd_modify_time
        |from
        |t5
        |""".stripMargin)





      //  关闭sparksql环境
    spark.close()
  }

}
